package cn.genmer.test.security.rpc.v2.rpc.transport;

import cn.genmer.test.security.rpc.SerialUtil;
import cn.genmer.test.security.rpc.v2.rpc.ResponseMappingCallback;
import cn.genmer.test.security.rpc.v2.rpc.protocol.GContent;
import cn.genmer.test.security.rpc.v2.rpc.protocol.GHeader;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;

import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

// 源于 spark源码
public class GClientFactory {
    int poolSize = 3;
    private static String PROTOCOL_TYPE_RPC = "RPC";
    private static String PROTOCOL_TYPE_HTTP = "HTTP";
    NioEventLoopGroup clientWorker;
    private static final GClientFactory factory;
    Random random = new Random();
    private static NioEventLoopGroup group = new NioEventLoopGroup(1); // 独立出去，res不知道怎么传

    static {
        factory = new GClientFactory();
    }
    private GClientFactory(){}
    public static GClientFactory getFactory(){
        return factory;
    }

    /**
     * 传输
     * @param content
     * @return
     */
    public static CompletableFuture  transport(String protocolType,GContent content){
        // content -> byteBuffer content就是货物，现在可以用自定义的rpc传输协议（有状态），也可以用http协议作为载体传输
        // 先手工使用http协议作为载体，那是不是代表我们未来可以让provider端是一个tomcat jetty 基于http协议的一个容器
        // 有无状态来自于使用的协议，HTTP协议无状态，每个请求对应一个连接
        // dubbo是一个rpc框架， netty是一个io框架
        // dubbo中传输协议上，可以是自定义的rpc传输协议，http协议
        CompletableFuture<Object> res = new CompletableFuture<>();
        if (PROTOCOL_TYPE_RPC.equals(protocolType)){
            byte[] msgBody = SerialUtil.serialObject(content);

            // 2. requestId+message,本地缓存 协议（Head+Body） = 【可参考Dubbo header<>】【msgBody】
            GHeader header = GHeader.createHeader(msgBody);
            byte[] msgHeader = SerialUtil.serialObject(header);

            // 3. 连接池：：取得连接
            NioSocketChannel clientChannel = factory.getClient(new InetSocketAddress("127.0.0.1", 7777));

            // 4.发送--> 走IO  out -->走Netty（event 驱动）
            ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgHeader.length + msgBody.length);

            // 获取requestID，设置回调
            long requestId = header.getRequestId();
            ResponseMappingCallback.addCallBack(requestId, res);

            byteBuf.writeBytes(msgHeader);
            byteBuf.writeBytes(msgBody);
            clientChannel.writeAndFlush(byteBuf);
        } else if (PROTOCOL_TYPE_HTTP.equals(protocolType)){
            // 1.使用URL现成的工具（包含了http编解码，发送，socket，连接）
//            urlTS(content,res);

            // 2. 自己操心：on netty(IO框架) + 已经提供的http相关编解码
            nettyTS(content, res);
        }


        return res;
    }

    /**
     * 自己使用netty来做IO相关操作+HTTP编解码
     * @param content
     * @param res
     */
    private static void nettyTS(GContent content, CompletableFuture<Object> res) {
        // 在这个执行之前， Server端 provider端已经开发完了，已经是on netty的http server了
        // 这个方法相较于 urlTS ，做的是consumer端的代码修改，改成on netty的http client（urlTS用的是JDK自带的）

        // 每个请求一个连接
        // 接下来的步骤
        // 1.通过netty建立IO 建立连接
        Bootstrap bs = new Bootstrap();
        Bootstrap socketClient = bs.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() { // 未来连接后，收到数据的处理handler
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline p = ch.pipeline();
                p.addLast(new HttpClientCodec())
                        .addLast(new HttpObjectAggregator(1024 * 512))
                        .addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                // 3. 接收 预埋的回调，根据netty对socket io事件的响应
                                FullHttpResponse response = (FullHttpResponse)msg;
//                                System.out.println("【客户端收到响应】：" + response.toString());

                                GContent responseContent = SerialUtil.parseGContent(response.content());
                                // 4.返回
                                res.complete(responseContent.getRes());
                            }
                        });
            }
        });

        // 2.发送
       try{
           ChannelFuture syncFuture = socketClient.connect("localhost", 7777).sync();
           Channel clientChannel = syncFuture.channel();
           byte[] data = SerialUtil.serialObject(content);
           DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_0, HttpMethod.POST, "/", Unpooled.copiedBuffer(data));
           // 不设置服务都读取会报content-length: 0 EOF
           request.headers().set(HttpHeaderNames.CONTENT_LENGTH, data.length);
           clientChannel.writeAndFlush(request).sync(); // 作为client 向 server端发送http request

        } catch (Exception e){
           System.out.println("客户端发送出现异常！");
           e.printStackTrace();
       }


    }

    /**
     * 使用现成的HTTP工具（包含了http编解码，发送，socket，连接）
     * @param content
     * @param res
     */
    private static void urlTS(GContent content, CompletableFuture<Object> res) {

        // 这种方式没请求占用一个连接，因为使用的是HTTP协议
        Object obj = null;
        try {
            URL url = new URL("http://localhost:7777/");
            HttpURLConnection connection =(HttpURLConnection) url.openConnection();

            connection.setRequestMethod("POST");
            connection.setDoOutput(true);
            connection.setDoInput(true);

            OutputStream outputStream = connection.getOutputStream();
            ObjectOutputStream objectOutputStream = new ObjectOutputStream(outputStream);
            objectOutputStream.writeObject(content);

            if (connection.getResponseCode() == 200){
                InputStream inputStream = connection.getInputStream();
                ObjectInputStream objectInputStream = new ObjectInputStream(inputStream);
                GContent gContent = (GContent) objectInputStream.readObject();
                obj = gContent.getRes();
            }


        }catch (Exception e){
            e.printStackTrace();
        }
        res.complete(obj);
    }

    // TODO 并发情况一定要谨慎
    //  一个consumer 可以连接很多的provider，每-个provider都有自己的pool K,V
    ConcurrentHashMap<InetSocketAddress, GClientPool> outboxs = new ConcurrentHashMap<>();

    public NioSocketChannel getClient(InetSocketAddress address){
        GClientPool clientPool = outboxs.get(address);
        // 还未初始化
        if (clientPool == null){
           synchronized (outboxs){
               // 如果之前clientPool指向为null，需要再次获取（因为不是基本数据类型）
             if (outboxs.get(address) == null){
                 outboxs.putIfAbsent(address, new GClientPool(poolSize));
                 clientPool = outboxs.get(address);
             }
           }
        }

        int i = random.nextInt(poolSize);
        Object lock = clientPool.locks[i];
        if (clientPool.clients[i] != null && clientPool.clients[i].isActive()){
            return clientPool.clients[i];
        }else {
            synchronized (lock){
                if (clientPool.clients[i] == null || Boolean.FALSE.equals(clientPool.clients[i].isActive()))
                    // 新建客户端需要使用数组下标
                clientPool.clients[i] = createClient(address);
            }
        }
        return clientPool.clients[i];
    }

    private NioSocketChannel createClient(InetSocketAddress address){
        // 基于Netty客户端 创建方式(无论多少线程，都是注册到这一个group)
         clientWorker = new NioEventLoopGroup(1);
        Bootstrap bs = new Bootstrap();
        ChannelFuture connect = bs.group(clientWorker)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {

                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        ChannelPipeline pipeline = nioSocketChannel.pipeline();
                        pipeline.addLast(new GCommonDecode());
                        pipeline.addLast(new GClientResponse()); // 通信上，解决response返回给谁
                    }
                }).connect(address);
       try {
           NioSocketChannel client = (NioSocketChannel) connect.sync().channel();
           return client;
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
        return null;
    }
}